package com.tc.rocketmq.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Auther: tianchao
 * @Date: 2021/9/5 20:55
 * @Description:
 */
// Define transaction listener with the annotation @RocketMQTransactionListener
@RocketMQTransactionListener(txProducerGroup = "rocketmq_test_boot_tran")
@Slf4j
public class MyTransactionImpl implements RocketMQLocalTransactionListener{
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<Object,Message> localTranMap = new ConcurrentHashMap<>();
    /**
     * 执行本地事务
     * @param msg
     * @param arg
     * @return 3426968b-1ec4-fa55-814f-c5d4e718c9ee
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.info("执行本地事务 {}",msg + "    " +arg);
        String tranId = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID).toString();
        log.error("本地事务id: {}", tranId);
        String tags = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS).toString();
        localTranMap.put(tranId,msg);
        if ("TagA".equals(tags)){
            return RocketMQLocalTransactionState.COMMIT;
        }else if ("TagB".equals(tags)){
            return RocketMQLocalTransactionState.ROLLBACK;
        }else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }

    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("检查本地事务 {}",msg);
        String transId = msg.getHeaders().get(RocketMQHeaders.PREFIX+RocketMQHeaders.TRANSACTION_ID).toString();
        log.error("检查事务id: {}", transId);
        Message message = localTranMap.get(transId);
        System.out.println("原始消息::::"+ message);
        String tags = msg.getHeaders().get(RocketMQHeaders.PREFIX + RocketMQHeaders.TAGS).toString();
        if ("TagC".equals(tags)){
            return RocketMQLocalTransactionState.COMMIT;
        }else if ("TagD".equals(tags)){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        if ("TagE".equals(tags)){
            System.out.println("第几次TagE来了:"+transactionIndex.incrementAndGet());
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        //"TagE...."
        return RocketMQLocalTransactionState.COMMIT;
    }
}
